package data

import (
	"context"
	"encoding/json"
	"github.com/gomodule/redigo/redis"
	"github.com/zhenjl/cityhash"
	"go_private_im/internal/logic/biz"
	"strconv"
)

// KeysByUserIds get a key server by userId.
func (l *LogicRepo) KeysByUserIds(c context.Context, userIds []int64) (ress map[string]string, oldUserIds []int64, err error) {
	conn := l.data.redis.Get()
	defer conn.Close()
	ress = make(map[string]string)
	for _, userId := range userIds {
		if err = conn.Send("HGETALL", keyUserIdServer(userId)); err != nil {
			l.log.Errorf("conn.Do(HGETALL %d) error(%v)", userId, err)
			return
		}
	}
	if err = conn.Flush(); err != nil {
		l.log.Errorf("conn.Flush() error(%v)", err)
		return
	}
	for idx := 0; idx < len(userIds); idx++ {
		var (
			res map[string]string
		)
		if res, err = redis.StringMap(conn.Receive()); err != nil {
			l.log.Errorf("conn.Receive() error(%v)", err)
			return
		}
		if len(res) > 0 {
			oldUserIds = append(oldUserIds, userIds[idx])
		}
		for k, v := range res {
			ress[k] = v
		}
	}
	return
}

// ServersByKeys get a server by key.
func (l *LogicRepo) ServersByKeys(c context.Context, keys []string) (res []string, err error) {
	conn := l.data.redis.Get()
	defer conn.Close()
	var args []interface{}
	for _, key := range keys {
		args = append(args, keyKeyServer(key))
	}
	if res, err = redis.Strings(conn.Do("MGET", args...)); err != nil {
		l.log.Errorf("conn.Do(MGET %v) error(%v)", args, err)
	}
	return
}
func (l *LogicRepo) AddMapping(ctx context.Context,userId int64,key,server string)  error {
	return  l.AddMappingByUserId(ctx,userId,key,server)
}

func (l *LogicRepo) Disconnect(ctx context.Context,req *biz.DisconnectReq) (has bool,err error) {
	conn := l.data.redis.Get()
	defer conn.Close()
	n := 1
	if req.UserId != 0 {
		if err = conn.Send("HDEL", keyUserIdServer(req.UserId), req.Key); err != nil {
			l.log.Errorf("conn.Send(HDEL %d,%s,%s) error(%v)", req.UserId, req.Key, req.Server, err)
			return
		}
		n++
	}
	if err = conn.Send("DEL", keyKeyServer(req.Key)); err != nil {
		l.log.Errorf("conn.Send(HDEL %d,%s,%s) error(%v)", req.UserId, req.Key, req.Server, err)
		return
	}
	if err = conn.Flush(); err != nil {
		l.log.Errorf("conn.Flush() error(%v)", err)
		return
	}
	for i := 0; i < n; i++ {
		if has, err = redis.Bool(conn.Receive()); err != nil {
			l.log.Errorf("conn.Receive() error(%v)", err)
			return
		}
	}
	return
}

func (l *LogicRepo) Heartbeat(ctx context.Context,req *biz.DisconnectReq) (err error) {
	conn := l.data.redis.Get()
	defer conn.Close()
	has, err := l.ExpireMapping(ctx, req.UserId, req.Key)
	if err != nil {
		l.log.Errorf("l.dao.ExpireMapping(%d,%s,%s) error(%v)", req.UserId, req.Key, req.Server, err)
		return
	}
	if !has {
		if err = l.AddMappingByUserId(ctx, req.UserId, req.Key, req.Server); err != nil {
			l.log.Errorf("l.dao.AddMapping(%d,%s,%s) error(%v)", req.UserId, req.Key, req.Server, err)
			return
		}
	}
	l.log.Infof("conn heartbeat key:%s server:%s userId:%d", req.Key, req.Server, req.UserId)
	return
}

func (l *LogicRepo) AddServerOnline(ctx context.Context,req *biz.Online)(map[string]int32,error)  {
	var err error
	roomsMap := map[uint32]map[string]int32{}
	for room, count := range req.RoomCount {
		rMap := roomsMap[cityhash.CityHash32([]byte(room), uint32(len(room)))%64]
		if rMap == nil {
			rMap = make(map[string]int32)
			roomsMap[cityhash.CityHash32([]byte(room), uint32(len(room)))%64] = rMap
		}
		rMap[room] = count
	}
	key := keyServerOnline(req.Server)
	for hashKey, value := range roomsMap {
		err = l.addServerOnline(ctx, key, strconv.FormatInt(int64(hashKey), 10), &biz.Online{RoomCount: value, Server: req.Server, Updated: req.Updated})
		if err != nil {
			return nil,err
		}
	}
	return l.roomCount,nil
}

func (l *LogicRepo) addServerOnline(c context.Context, key string, hashKey string, online *biz.Online) (err error) {
	conn := l.data.redis.Get()
	defer conn.Close()
	b, _ := json.Marshal(online)
	if err = conn.Send("HSET", key, hashKey, b); err != nil {
		l.log.Errorf("conn.Send(SET %s,%s) error(%v)", key, hashKey, err)
		return
	}
	if err = conn.Send("EXPIRE", key, l.redisExpire); err != nil {
		l.log.Errorf("conn.Send(EXPIRE %s) error(%v)", key, err)
		return
	}
	if err = conn.Flush(); err != nil {
		l.log.Errorf("conn.Flush() error(%v)", err)
		return
	}
	for i := 0; i < 2; i++ {
		if _, err = conn.Receive(); err != nil {
			l.log.Errorf("conn.Receive() error(%v)", err)
			return
		}
	}
	return
}

// ExpireMapping expire a mapping.
func (l *LogicRepo) ExpireMapping(c context.Context, userId int64, key string) (has bool, err error) {
	conn := l.data.redis.Get()
	defer conn.Close()
	var n = 1
	if userId != 0 {
		if err = conn.Send("EXPIRE", keyUserIdServer(userId), l.redisExpire); err != nil {
			l.log.Errorf("conn.EXPIRE(HDEL %d,%s) error(%v)",userId, key, err, err)
			return
		}
		n++
	}
	if err = conn.Send("EXPIRE", keyKeyServer(key), l.redisExpire); err != nil {
		l.log.Errorf("conn.Send(EXPIRE %d,%s) error(%v)", userId, key, err)
		return
	}
	if err = conn.Flush(); err != nil {
		l.log.Errorf("conn.Flush() error(%v)", err)
		return
	}
	for i := 0; i < n; i++ {
		if has, err = redis.Bool(conn.Receive()); err != nil {
			l.log.Errorf("conn.Receive() error(%v)", err)
			return
		}
	}
	return
}

func (l *LogicRepo) AddMappingByUserId (c context.Context,userId int64, key string,server string)  (err error) {
	conn := l.data.redis.Get()
	defer conn.Close()
	var n = 2
	if userId != 0 {
		if err = conn.Send("HSET", keyUserIdServer(userId),key, server); err != nil {
			l.log.Errorf("conn.Send(HSET %d,%s,%s) error(%v)", userId, server, key, err)
			return
		}
		if err = conn.Send("EXPIRE", keyUserIdServer(userId), l.redisExpire); err != nil {
			l.log.Errorf("conn.Send(EXPIRE %d,%s,%s) error(%v)",  userId, server, key, err)
			return
		}
		n += 2
	}
	if err = conn.Send("SET", keyKeyServer(key), server); err != nil {
		l.log.Errorf("conn.Send(HSET %d,%s,%s) error(%v)",  userId, server, key, err)
		return
	}
	if err = conn.Send("EXPIRE", keyKeyServer(key), l.redisExpire); err != nil {
		l.log.Errorf("conn.Send(EXPIRE %d,%s,%s) error(%v)",  userId, server, key, err)
		return
	}
	if err = conn.Flush(); err != nil {
		l.log.Error(err.Error())
		return
	}
	for i := 0; i < n; i++ {
		if _, err = conn.Receive(); err != nil {
			l.log.Errorf("conn.Receive() error(%v)", err)
			return
		}
	}
	return
}


// ServerOnline get a server online.
func (l *LogicRepo) ServerOnline(c context.Context, server string) (online *biz.Online, err error) {
	online = &biz.Online{RoomCount: map[string]int32{}}
	key := keyServerOnline(server)
	for i := 0; i < 64; i++ {
		ol, err := l.serverOnline(c, key, strconv.FormatInt(int64(i), 10))
		if err == nil && ol != nil {
			online.Server = ol.Server
			if ol.Updated > online.Updated {
				online.Updated = ol.Updated
			}
			for room, count := range ol.RoomCount {
				online.RoomCount[room] = count
			}
		}
	}
	return
}

func (l *LogicRepo) serverOnline(c context.Context, key string, hashKey string) (online *biz.Online, err error) {
	conn := l.data.redis.Get()
	defer conn.Close()
	b, err := redis.Bytes(conn.Do("HGET", key, hashKey))
	if err != nil {
		if err != redis.ErrNil {
			l.log.Errorf("conn.Do(HGET %s %s) error(%v)", key, hashKey, err)
		}
		return
	}
	online = new(biz.Online)
	if err = json.Unmarshal(b, online); err != nil {
		l.log.Errorf("serverOnline json.Unmarshal(%s) error(%v)", b, err)
		return
	}
	return
}

// DelServerOnline del a server online.
func (l *LogicRepo) DelServerOnline(c context.Context, server string) (err error) {
	conn := l.data.redis.Get()
	defer conn.Close()
	key := keyServerOnline(server)
	if _, err = conn.Do("DEL", key); err != nil {
		l.log.Errorf("conn.Do(DEL %s) error(%v)", key, err)
	}
	return
}
